//SPDX-FileCopyrightText: Copyright 2022-2024 深圳市同心圆网络有限公司
//SPDX-License-Identifier: GPL-3.0-only

package user_api_serv

import (
	"context"
	"strings"
	"time"

	"atomgit.com/openlinksaas/api-server/api_common"
	"atomgit.com/openlinksaas/api-server/dao"
	"atomgit.com/openlinksaas/api-server/dao/project_dao"
	"atomgit.com/openlinksaas/api-server/notice_serv"
	"atomgit.com/openlinksaas/proto-gen-go.git/notices/notices_user"
	"atomgit.com/openlinksaas/proto-gen-go.git/user_api"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
)

func watchUser() {
	msgChan := dao.CacheDao.GetMsgChan()
	for msg := range msgChan {
		if strings.HasPrefix(msg.Payload, dao.CACHE_PREFIX_USER) {
			userId := msg.Payload[len(dao.CACHE_PREFIX_USER):]
			sendOfflineNotice(userId)
		}
	}
}

func sendOfflineNotice(userId string) {
	notice := notices_user.UserOfflineNotice{
		UserId: userId,
	}
	sendNotice(userId, &notice)
}

func sendOnlineNotice(userId string) {
	notice := notices_user.UserOnlineNotice{
		UserId: userId,
	}
	sendNotice(userId, &notice)
}

func sendNotice(userId string, notice proto.Message) {
	projectIdList, err := project_dao.MemberInfoDao.ListProjectByMember(context.Background(), userId)
	if err != nil {
		return
	}
	memberItemList, err := project_dao.MemberInfoDao.ListMemberByProject(context.Background(), projectIdList)
	if err != nil {
		return
	}
	tmpMap := map[string]bool{}
	for _, memberItem := range memberItemList {
		tmpMap[memberItem.MemberUserId] = true
	}
	for memberUserId := range tmpMap {
		go api_common.SendNoticeToUser(context.Background(), memberUserId, notice)
	}
}

func (apiImpl *UserApiImpl) PullNotice(req *user_api.PullNoticeRequest, s user_api.UserApi_PullNoticeServer) error {
	ctx := s.Context()
	sessItem, ok := dao.CacheDao.GetSession(ctx, req.SessionId)
	if !ok {
		return s.Send(&user_api.PullNoticeResponse{
			Code:       user_api.PullNoticeResponse_CODE_WRONG_SESSION,
			ErrMsg:     "会话不存在",
			NoticeData: &anypb.Any{},
		})
	}
	subCh, err := notice_serv.SubscribeToUser(ctx, sessItem.UserId)
	if err != nil {
		return err
	}
	defer subCh.Close()

	msgCh := subCh.Channel()
	for {
		select {
		case msg := <-msgCh:
			notice := anypb.Any{}
			err = proto.Unmarshal([]byte(msg.Payload), &notice)
			if err != nil {
				continue
			}
			err = s.Send(&user_api.PullNoticeResponse{
				Code:       user_api.PullNoticeResponse_CODE_OK,
				NoticeData: &notice,
			})
			if err != nil {
				return err
			}
		case <-time.After(30 * time.Second):
			_, ok = dao.CacheDao.GetSession(ctx, req.SessionId)
			if !ok {
				return s.Send(&user_api.PullNoticeResponse{
					Code:       user_api.PullNoticeResponse_CODE_WRONG_SESSION,
					ErrMsg:     "会话不存在",
					NoticeData: &anypb.Any{},
				})
			}
		}

	}
}
